package ApplicationTest.Example.KafKa.Producer

import kafka.serializer.StringEncoder
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
import org.codehaus.jettison.json.JSONObject

import scala.util.Random

object KafkaProducer {


  private val users = Array(
    "4A4D769EB9679C054DE81B973ED5D768", "8dfeb5aaafc027d89349ac9a20b3930f",
    "011BBF43B89BFBF266C865DF0397AA71", "f2a8474bf7bd94f0aabbd4cdd2c06dcf",
    "068b746ed4620d25e26055a9f804385f", "97edfc08311c70143401745a03a50706",
    "d7f141563005d1b5d0d3dd30138f3f62", "c8ee90aade1671a21336c721512b817a",
    "6b67c8c700427dee7552f81f3228c927", "a95f22eabc4fd4b580c011a3161a9d9d")

  private val random = new Random() //create 随机数

  private var pointer = -1 //默认指针

  def getUserID: String = {
    pointer = pointer + 1
    if(pointer >= users.length) {
      pointer = 0
      users(pointer)
    } else {
      users(pointer)
    }
  }

  def click : Double = {
    random.nextInt(10)
  }

  // bin/kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181/kafka --create --topic user_events --replication-factor 2 --partitions 2
  // bin/kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181/kafka --list
  // bin/kafka-topics.sh --zookeeper zk1:2181,zk2:2181,zk3:2181/kafka --describe user_events
  // bin/kafka-console-consumer.sh --zookeeper zk1:2181,zk2:2181,zk3:22181/kafka --topic test_json_basis_event --from-beginning

  def main(args: Array[String]): Unit = {



    val topic = Array("helloword")
    val brokers = "master:9092,spark02:9092"
    val bootstrapServer = "master:9092,spark02:9092,spark03:9092"

    import scala.collection.JavaConversions._

    val propsMap : Map[String, Object] = Map(
      "metadata.broker.list"-> brokers,
      "bootstrap.servers" -> bootstrapServer,
      "serializer.class" -> classOf[StringEncoder],
      "key.serializer" -> classOf[StringSerializer],
      "value.serializer" -> classOf[StringSerializer],
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val producer : KafkaProducer[String, String] = new KafkaProducer(propsMap)

    while(true) {
      // prepare event data
      val event = new JSONObject()
      event
        .put("uid", getUserID)
        .put("event_time", System.currentTimeMillis.toString)
        .put("os_type", "Android")
        .put("click_count", click)

      // produce event message
      val producerRecord = new ProducerRecord[String, String](topic(0), "key:" + click.toString, event.toString)
      //send information
      producer.send(producerRecord)

      println("Message sent: " + event)

      Thread.sleep(1500)
    }
  }
}
